package com.atguigu.gmall.realtime.app.func;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.common.GmallConfig;
import com.atguigu.gmall.realtime.utils.DimUtil;
import com.atguigu.gmall.realtime.utils.PhoenixUtil;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;

import java.util.Map;

public class DimSinkFunction implements SinkFunction<JSONObject> {
    @Override
    public void invoke(JSONObject jsonObj , Context context) throws Exception {
        //1  把jsonobj中的数据提取转换为sql语句
        // upsert  into  库名.表名(字段名)   values(字段值 )  ;
        String tableName=jsonObj.getString("sink_table");
        StringBuilder colNameSB=new StringBuilder();
        StringBuilder colValueSB=new StringBuilder();
        JSONObject dataJsonObj = jsonObj.getJSONObject("data");
        for (Map.Entry<String, Object> entry : dataJsonObj.entrySet()) {
            if(colNameSB.length()>0){
                colNameSB.append( ",");
                colValueSB.append( ",");
            }
            colNameSB.append( entry.getKey());
            if(entry.getValue()==null){
                colValueSB.append("null");
            }else{
                colValueSB.append("'"+entry.getValue()+"'");
            }
        }

        String upsertSQL="upsert into "+ GmallConfig.DIM_SCHEMA+"."+tableName
                +"("+colNameSB+") values( " +colValueSB+")";
        //2执行
        System.out.println("upsert sql :"+upsertSQL);
        PhoenixUtil.executeSQL(upsertSQL);

        //如果维度数据发生了变化，将Redis中缓存的维度数据删除掉
        String type = jsonObj.getString("type");
        if("update".equals(type)){
            DimUtil.deleteCached(tableName,dataJsonObj.getString("id"));
        }
    }

}
